package data.basic

import data.bean.Domain.WC
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.Table
import org.apache.flink.types.Row

// 开发时要做的第一件事情
import org.apache.flink.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._

object DataStreamTableSQLApp03 {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnvironment = StreamTableEnvironment.create(env)


    val stream = env.fromElements("andy,andy,andy", "sb,sb", "dingding")
      .flatMap(_.split(",")).map(WC(_))

    // datastream ==> table
    val accessTable: Table = tableEnvironment.fromDataStream(stream)
    tableEnvironment.sqlQuery(
      s"""
        |
        |select
        |word, count(word)
        |from $accessTable
        |group by word
        |""".stripMargin)
      .toRetractStream[Row].filter(_._1).print("SQL写法...")


    tableEnvironment.fromDataStream(stream, 'word)
        .groupBy('word)
        .select('word, 'word.count())
      .toRetractStream[Row].filter(_._1).printToErr("API写法...")


    env.execute(getClass.getCanonicalName)
  }
}
